package roohom;

import com.changan.common.model.Tuple2;
import com.changan.config.ColumnEntity;
import com.changan.config.TableEntity;
import com.changan.model.AppArgs;
import com.changan.parser.CliTool;
import com.changan.services.FlinkService;
import com.changan.utils.MysqlUtil;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;

import static com.changan.StreamApp.insertSql;

public class InitJobSql {


    private static final String CDC_PREFIX = "cdc_";
    private static final String STARROCKS_PREFIX = "starrocks_";


    public static void main(String[] args) {
        //-jn flink-starrocks-cdc -sn mysql -st user_info_data -spk id -tn starrocks -tt user_info_data -tpk id -kf kafka -kt star_user_info_data_topic -ka cdh001:9092,cdh002:9092,cdh003:9092 -kp id -kts _topic -kgp flink-cdc-starrocks -kssm earliest-offset -sa jdbc:mysql://192.168.2.45:3306/star -un root -pw zhongdiao7115240@ -hs 192.168.2.45 -pt 3306 -db star -ta jdbc-url=jdbc:mysql://192.168.2.43:9030|load-url=192.168.2.43:8030 -tu root -tpw zd1qaz@WSX -tdb star -sum initial
        //-jn flink-starrocks-cdc -sn mysql -st new_tabe -spk id -tn starrocks -tt new_tabe -tpk id -kf kafka -kt star_new_table_topic -ka cdh001:9092,cdh002:9092,cdh003:9092 -kp id -kts _topic -kgp flink-cdc-starrocks -kssm earliest-offset -sa jdbc:mysql://192.168.2.47:13306/star -un root -pw zd123456 -hs 192.168.2.47 -pt 13306 -db star -ta jdbc-url=jdbc:mysql://192.168.2.43:9030|load-url=192.168.2.43:8030 -tu root -tpw zd1qaz@WSX -tdb star -sum initial

        AppArgs appArgs = CliTool.getJcommander(AppArgs.class, args);
        System.out.printf(String.valueOf(appArgs));

        List<Tuple2<TableEntity, List<ColumnEntity>>> sources = FlinkService.getFlinkTaskSourceDefinitions(appArgs);
        List<Tuple2<TableEntity, List<ColumnEntity>>> targets = FlinkService.getFlinkTaskTargetDefinitions(appArgs, sources);

        for (int i = 0; i < sources.size(); i++) {
            Tuple2<TableEntity, List<ColumnEntity>> source = sources.get(i);
            Tuple2<TableEntity, List<ColumnEntity>> target = targets.get(i);

            String sourceViewName = CDC_PREFIX + source._1.getQualifier();
            String kafkaSql = FlinkService.getDebeziumSql(appArgs, source._1, source._2, sourceViewName);
            System.out.println(kafkaSql);

            String sinkViewName = STARROCKS_PREFIX + target._1.getQualifier();
            String starrocksSql = FlinkService.getStarrocksSql(appArgs, target._1, target._2, sinkViewName);
            System.out.println(starrocksSql);

            String insertSql = insertSql(sourceViewName, sinkViewName);
            System.out.println(insertSql);
            System.out.println("=============================================");
        }
    }

    public static LinkedHashMap<String, List<ColumnEntity>> parseColumnsFromJdbc(String url, String userName, String password, String databaseName, String... tableName) throws SQLException {
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }

        LinkedHashMap<String, List<ColumnEntity>> columnMaps = new LinkedHashMap<>();

        Connection connection = MysqlUtil.getConnection(url, userName, password);

        Statement statement = connection.createStatement();

        for (String table : tableName) {
            String describe = "describe " + table;
            System.out.println(describe);
            ResultSet resultSet = statement.executeQuery(describe);
            ArrayList<ColumnEntity> columnEntities = new ArrayList<>();
            int i = 1;
            while (resultSet.next()) {
                String fieldName = resultSet.getString(1);
                String type = resultSet.getString(2);
                String typeWithPrecision = type.split("\\s+")[0];
                String actualTypeName = typeWithPrecision.contains("(") ? typeWithPrecision.substring(0, typeWithPrecision.indexOf("(")) : typeWithPrecision;

                //System.out.println(fieldType.getFlinkType());
                String nullAble = resultSet.getString(3);
                String isKey = resultSet.getString(4);
                String defaultValue = resultSet.getString(5);
                String extra = resultSet.getString(6);

                ColumnEntity columnEntity = new ColumnEntity();
                columnEntity.setDatabaseName(databaseName);
                columnEntity.setTableName(table);
                columnEntity.setColName(fieldName);
                columnEntity.setFlinkColName(fieldName);
                columnEntity.setFlinkColType(actualTypeName);
                columnEntity.setColType(typeWithPrecision);
                columnEntity.setColPosition(i);
                columnEntity.setIsPrimaryKey("PRI".equals(isKey));

                columnEntities.add(columnEntity);
                i++;
            }
            columnMaps.put(table, columnEntities);
        }

        MysqlUtil.closeConnection(connection);

        return columnMaps;
    }
}
